/*
* 作者: wupengxin
* 创建时间: 2016-08-16 14:42:50
* 说明: 读取文件内容的插件
 */
package inputplug

import (
	"bufio"
	"common"
	//"encoding/json"
	"errors"
	. "fmt"
	"github.com/fsnotify/fsnotify"
	"io"
	"log"
	"os"
	"path/filepath"
	"regexp"
	"strconv"
	"strings"
	"time"
	"sync"
)

type FileReader struct {
	Path         []string       // 待读取的文件路径
	Files        []string       // 待读取的文件
	Exclude      []string       // 指定某些文件名不读取
	NewfileTime  int            // 发现新文件的间隔时间 单位为S
	FileChange   int            //检测文件改变的时间
	ReadPosition string         //每次启动之后 读取文件的位置
	SyncPath     string         // 保存读取文件的位移信息
	Pattern      *regexp.Regexp // 正则匹配 将匹配的内容作为一个整体输出
	Channel      chan map[string]string
}

// 保存文件事件
var sg_file_env = make(map[string]chan string)

var sg_lock sync.RWMutex

//文件读取后输出的格式
type outJson struct {
	Message   string `json:"message"`
	Timestamp string `json:"timestamp"`
	File      string `json:"file"`
}

//递归遍历文件下的所有文件
func walkAllfiles(root string, list []string) ([]string, error) {
	err := filepath.Walk(root, func(path string, fi os.FileInfo, err error) error {
		if nil == fi {
			return err
		}
		if fi.IsDir() {
			return nil
		}
		list = append(list, path)

		return nil
	})

	if err != nil {
		log.Println(time.Now(), err)
	}
	return list, err
}

// 遍历文件夹下的所有子文件夹


/*
* 功能:获取指定路径下的文件名称
* 说明:  比如获取/var/log/目录下的所有文件文件
* TODO:: 需要支持正则表达式
 */

func (fileread *FileReader) GetPathFile() ([]string, error) {
	filelist := make([]string, 0)
	for _, path := range fileread.Path {
		// 获取文件的绝对路径
		absPath, _ := filepath.Abs(path)
		// 判断输入的路径是文件 还是文件夹
		if _, e := os.Stat(absPath); e != nil {
			//说明该文件或者文件夹不存在
			common.Debug(e)
			return nil, e
		} else {
				filelist, _ = walkAllfiles(absPath, filelist)
			}
			continue
		}

	return filelist, nil
}

// 根据设置的指定模式来读取文件 以行为单位进行读取,以指定的模式进行输出到通道
func (fileread *FileReader) ReadFileByLine(path string, ch chan map[string]string) error {
	//var buffer bytes.Buffer
	f, e := os.Open(path)
	var appid, container_id, file string
	// 获取appid container_id
	slice := strings.Split(path, "/")
	file = slice[len(slice)-1]
	if len(slice) < 3{
		appid = "unkown"
		container_id = "unkown"
	}else {
		appid = slice[len(slice)-3]
		container_id = slice[len(slice)-2]
	}

	if e != nil {
		common.Debug(e)
		return e
	}
	// 异常捕获
	defer func() {
		//close(ch)
		f.Close()
		if err := recover(); err != nil {
			common.Debug(err) //
		}
		common.Debug("read file worker exit...")
	}()

	bfRd := bufio.NewReader(f)
	var read string
	for {
		line, err := bfRd.ReadString('\n')
		if err != nil {
			if err == io.EOF {
				//读到文件结尾 则一直等待有新的内容传入
				if channel, ok := sg_file_env[path]; ok{
					contant := <- channel
					if contant == "readable"{
						common.Debug("file readable:" + path)
						continue
					} else if contant == "delete"{
						delete(sg_file_env, path)
						common.Debug("delet file:" + path)
						return nil
					}
				}

				if _, e := os.Stat(path); e != nil {
					//说明该文件或者文件夹不存在
					return e
				}
				common.Debug(path+"file end------")

				continue
			}
			return err
		}
		read = Sprintf("%s%s", read, line)
		//buffer.WriteString(string(line))
		if len(read) > 1024*1024*10 {
			// 如果读取了100M 还没输出 就退出
			err = errors.New("已经读取了10M内容,依然匹配不成功.退出----")
			common.Debug(err)
			return err
		}

		//判断是否可以作为一个整体输出
		index := fileread.Pattern.FindStringIndex(read)
		if index == nil {
			//Println("未匹配成功", read)
			continue
		}
		//jsonStr := &outJson{Message: read[0 : index[0]+1], Timestamp: Sprintf("%s", time.Now()), File: path}
		msg := make(map[string]string)
		msg["contant"] = read[0 : index[0]+1]
		msg["appid"] = appid
		msg["container_id"] = container_id
		msg["file"] = file
		ch <- msg
		//if str, e := json.Marshal(jsonStr); e != nil {
		//	common.Debug(e)
		//	return e
		//} else {
		//	//Println("json格式内容: ", string(str))
		//	ch <- string(str)
		//}
		read = read[index[0]+1:]
	}
}

// 实现约定接口
func (my *FileReader) GetInputStream(conf map[string]string) ([]chan map[string]string, error) {
	//根据配置信息 初始化FileReader对象
	// 1.获取文件件路径
	my.Path = strings.Split(conf["path"], ":")
	// 2.根据路径获得文件列表
	var e error
	if my.Files, e = my.GetPathFile(); e != nil {
		common.Debug(e)
		return nil, e
	}
	// 3. 需要排除的文件
	my.Exclude = strings.Split(conf["exclude"], ":")
	// 4. 新文件的间隔时间
	my.NewfileTime, _ = strconv.Atoi(conf["newfiletime"])
	// 5.保存文件路径
	my.SyncPath = conf["syncpath"]
	// 6.多行匹配的正则
	if my.Pattern, e = regexp.Compile(conf["mattch"]); e != nil {
		common.Debug(e)
		return nil, e
	}
	/*
		// 为每一个文件生成一个通道 为每一个文件生成goroute 读取内容
		channels := make([]chan string, len(my.Files))
		// 注意上面只是初始化了一个nil channel 对nil进行读写都是会阻塞 现在需要进行创建
		for i := 0; i < len(my.Files); i++ {
			channels[i] = make(chan string)
		}
		for index, filename := range my.Files {
			go my.ReadFileByLine(filename, channels[index])
		}*/

	// 2016-10-02 17:30:12 修改内容传输模式 改成所有文件通过一个通道传输出去
	channels := make([]chan map[string]string, 1)
	channels[0] = make(chan map[string]string)
	my.Channel = channels[0]
	for _, filename := range my.Files {
		common.Debug("read file:" + filename)
		go my.ReadFileByLine(filename, channels[0])
	}
	return channels, nil
}

// 实现约定接口 设置配置信息
func (my *FileReader) ConfigPlus(conf interface{}) (map[string]string, error) {
	defer func() {
		if e := recover(); e != nil {
			common.Debug(e)
		}
	}()

	tmp := make(map[string]string)

	dat := conf.(map[string]interface{})
	var ok bool
	// 读取文件的路径
	if _, ok = dat["path"]; !ok {
		return nil, errors.New("input must config path")
	} else {
		tmp["path"] = dat["path"].(string)
	}
	// 不读取的文件
	if _, ok = dat["exclude"]; !ok {
		tmp["exclude"] = ""
	} else {
		tmp["exclude"] = dat["exclude"].(string)
	}
	// 每次检查新文件的时间
	if _, ok = dat["newfiletime"]; !ok {
		tmp["newfiletime"] = "10"
	} else {
		tmp["newfiletime"] = dat["newfiletime"].(string)
	}
	// 保存读取文件指针的位置
	if _, ok = dat["syncpath"]; !ok {
		tmp["syncpath"] = "./"
	} else {
		tmp["syncpath"] = dat["syncpath"].(string)
	}
	// 多行匹配的模式
	if _, ok = dat["mattch"]; !ok {
		return nil, errors.New("input must config mattch")
	} else {
		tmp["mattch"] = dat["mattch"].(string)
	}
	return tmp, nil
}


//递归监视当前文件夹下的所有文件夹
func recursiveAddWatch(root string, watch *fsnotify.Watcher) (error) {
	watch.Add(root)
	err := filepath.Walk(root, func(path string, fi os.FileInfo, err error) error {
		if nil == fi {
			return err
		}
		if fi.IsDir() {
			watch.Remove(path)
			watch.Add(path)
		} else {
			sg_lock.Lock()
			sg_file_env[path] = make(chan string, 3)
			sg_lock.Unlock()
		}
		return nil
	})

	return err
}

// 实现接口
func (my *FileReader) Start() {
	go func() {
		watcher, err := fsnotify.NewWatcher()
		if err != nil {
			common.Exit(errors.New("创建文件监视对象失败"))
		}
		defer watcher.Close()
		for _, path := range my.Path{
			err = recursiveAddWatch(path, watcher)
			if err != nil {
				common.Exit(err)
			}
		}

		// 开始实时监控文件变化
		for {
			select {
			case event := <-watcher.Events:
				if event.Op&fsnotify.Write == fsnotify.Write {
					if channel, ok := sg_file_env[event.Name]; ok{
						select {
						case channel <- "readable":
						default:
							break
						}

					}else{
						sg_lock.Lock()
						sg_file_env[event.Name] = make(chan string, 3)
						sg_lock.Unlock()
						select {
						case sg_file_env[event.Name] <- "readable":
						default:
							break
						}
					}
				}
				if event.Op&fsnotify.Create == fsnotify.Create{
					fi, _ := os.Stat(event.Name)
					if fi.IsDir(){
						watcher.Add(event.Name)
					}else {
						// 如果新添加的是文件 创建新的goroute
						sg_lock.Lock()
						sg_file_env[event.Name] = make(chan string, 3)
						sg_lock.Unlock()
						go my.ReadFileByLine(event.Name, my.Channel)
					}
				}
				if event.Op&fsnotify.Remove == fsnotify.Remove{
					watcher.Remove(event.Name)
					if channel, ok := sg_file_env[event.Name]; ok{
						select {
						case channel <- "delete":
						default:
							break
						}
					}
				}
			case err := <-watcher.Errors:
				common.Debug(err)
			}
		}
	}()

}
